package com.zyx.flinkdemo.stream.param;


import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author zyx
 * @since 2021/5/22 08:25
 * desc: 全局参数实例
 */
public class GlobalParamDemo {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setLong("limit", 16L);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().setGlobalJobParameters(conf);

        DataStreamSource<Long> longDs = env.fromSequence(1, 30);

        longDs.filter(new FilterJobParameters()).print("JobParameters stream is :");

        env.execute();
    }

    private static class FilterJobParameters extends RichFilterFunction<Long> {

        protected long limit;

        @Override
        public void open(Configuration parameters) {
            ExecutionConfig executionConfig = getRuntimeContext().getExecutionConfig();
            ExecutionConfig.GlobalJobParameters globalParams = executionConfig.getGlobalJobParameters();
            Configuration globConf = (Configuration) globalParams;
            ConfigOption<Long> limit = ConfigOptions.key("limit").longType().defaultValue(0L);
            this.limit = globConf.getLong(limit);
        }

        @Override
        public boolean filter(Long value) {
            return value > limit;
        }
    }
}
